package com.shujia.sql

import org.apache.spark.sql.{DataFrame, SparkSession}

object Code10DSL {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("dsl")
      .config("spark.sql.shuffle.partitions", "3")
      .getOrCreate()

    val stuDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      // 1500100001,施笑槐,22,女,文科六班
      .schema("id String,name String,age int,gender String,clazz String")
      .load("spark_code/data/students_join.txt")


    /**
     * DSL函数：
     * 使用前必须要导入  import org.apache.spark.sql.functions._
     */


    // sql函数
    import org.apache.spark.sql.functions._
    // Spark隐式转换
    import spark.implicits._

    val scoreDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      // 1500100001,施笑槐,22,女,文科六班
      .schema("id String,courseID String,score int")
      .load("spark_code/data/score_join.txt")

    stuDF
      .groupBy("clazz")
      .agg(collect_list("id") as "all_id")
      .select($"clazz", concat_ws("\t", $"all_id"))
      .show(truncate = false)


    stuDF
      .select( $"id",md5($"id") as "md5_id")
//      .where($"id" === md5($"'1500100001'"))
      .where(expr("md5_id = md5('1500100001')"))
      .show()


    /**
     * when函数使用：
     *    when(判断条件,结果为true返回值).when(判断条件,结果为true返回值).otherwise(所有为false的返回值)
     *
      */

    stuDF
//      .select($"id",$"name",$"age",$"gender",$"clazz",when($"age" > 22,1).when($"age" <= 22,0))
      .select($"id",$"name",$"age",$"gender",$"clazz",when($"age" > 22,1).otherwise(0) as "when_res")
      .show()

    /**
     * withColumn 可以给定列名及列表达式 添加一列数据
     * withColumnRenamed  对列名称进行重命名操作
     */
    stuDF
      .withColumn("when_res",when($"age" > 22,1).otherwise(0))
      .withColumnRenamed("when_res","new_when_res")
      .show()



  }
}
